package com.fwmagic.flinkcdc.schema;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

/**
 * @description: 自定义json反序列化类
 * @author: fangwei
 * @date: 2022/5/18
 **/
public class JsonDeserializationSchema implements DebeziumDeserializationSchema<JSONObject> {
    private static final long serialVersionUID = 4186899475568749993L;


    public JsonDeserializationSchema() {
    }

    @Override
    public void deserialize(SourceRecord record, Collector<JSONObject> out) throws Exception {
        Struct valueStruct = (Struct) record.value();
        Struct sourceStrut = valueStruct.getStruct("source");
        //获取数据库的名称
        String database = sourceStrut.getString("db");
        //获取表名
        String table = sourceStrut.getString("table");

        //获取类型（c-->insert,u-->update）
        String type = Envelope.operationFor(record).toString().toLowerCase();
        //read
        //System.out.println("type = "+type);
        if (type.equals("create")) {
            type = "insert";
        }


        JSONObject json = new JSONObject();
        json.put("database", database);
        json.put("table", table);
        json.put("type", type);

        //获取数据data
        Struct afterStruct = valueStruct.getStruct("after");
        setData(json, "after", afterStruct);

        Struct beforeStruct = valueStruct.getStruct("before");
        setData(json, "before", beforeStruct);

        //向下游传递数据
        out.collect(json);
    }

    private void setData(JSONObject json, String type, Struct struct) {
        JSONObject dataJson = new JSONObject();
        if (struct != null) {
            for (Field field : struct.schema().fields()) {
                String fieldName = field.name();
                Object fieldValue = struct.get(field);
                dataJson.put(fieldName, fieldValue);
            }
        }
        json.put(type, dataJson);
    }

    @Override
    public TypeInformation<JSONObject> getProducedType() {
        return BasicTypeInfo.of(JSONObject.class);
    }
}